package com.qqt.csr.web.timer;

import cn.hutool.core.date.DatePattern;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.thread.ThreadUtil;
import com.alibaba.fastjson2.JSONObject;
import com.google.common.base.Stopwatch;
import com.google.common.collect.Lists;
import com.qqt.csr.common.acl.aics.AICustomerServiceClient;
import com.qqt.csr.common.acl.aics.resp.CsMemberInfoDTO;
import com.qqt.csr.common.enums.StatusEnum;
import com.qqt.csr.common.utils.CacheKeyUtil;
import com.qqt.csr.common.utils.RedissonLockUtil;
import com.qqt.csr.common.utils.SnowflakeUtil;
import com.qqt.csr.common.utils.ThreadPoolUtil;
import com.qqt.csr.im.dto.SendSystemMessageDTO;
import com.qqt.csr.im.entity.Account;
import com.qqt.csr.im.entity.Conversation;
import com.qqt.csr.im.entity.msg.Message;
import com.qqt.csr.im.enums.ConversationStateEnum;
import com.qqt.csr.im.service.AccountService;
import com.qqt.csr.im.service.mongo.ConversationService;
import com.qqt.csr.im.service.mongo.MessageService;
import com.qqt.csr.route.service.ConversationStatService;
import com.qqt.csr.route.service.CsReceptionLimitService;
import com.qqt.csr.web.service.ArchiveEventMqProducerService;
import com.qqt.csr.web.service.ImAccountService;
import com.qqt.csr.workbench.entity.ChatTimeJob;
import com.qqt.csr.workbench.entity.SkillSetConfig;
import com.qqt.csr.workbench.service.ChatTimeJobService;
import com.qqt.csr.workbench.service.SkillSetConfigService;
import jakarta.annotation.PreDestroy;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.math.NumberUtils;
import org.apache.commons.lang3.time.DateUtils;
import org.slf4j.MDC;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import java.util.*;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

@Slf4j
@Service
public class ChatTimeJobTimer implements InitializingBean {
    private static final String LOG_TID = "ctid";
    private final ScheduledThreadPoolExecutor executor = ThreadUtil.createScheduledExecutor(1);
    private volatile boolean isShutdown = false;

    @Value(value = "${csr.chatTimeJob.interval}")
    private long interval;
    @Value(value = "${csr.chatTimeJob.limit}")
    private int limit;
    @Autowired
    private RedissonLockUtil redissonLockUtil;
    @Autowired
    private SkillSetConfigService skillSetConfigService;
    @Autowired
    private ConversationService conversationService;
    @Autowired
    private MessageService messageService;
    @Autowired
    private AccountService accountService;
    @Autowired
    private CsReceptionLimitService csReceptionLimitService;
    @Autowired
    private ArchiveEventMqProducerService archiveEventMqProducerService;
    @Autowired
    private AICustomerServiceClient aiCustomerServiceClient;
    @Autowired
    private ChatTimeJobService chatTimeJobService;
    @Autowired
    private ConversationStatService conversationStatService;
    @Autowired
    private ImAccountService imAccountService;

    @Override
    public void afterPropertiesSet() throws Exception {
        ThreadUtil.schedule(executor,
                () -> {
                    try {
                        MDC.put(LOG_TID, SnowflakeUtil.nextIdStr());
                        redissonLockUtil.distributedLock(this::monitorJob, CacheKeyUtil.SkillSet.CHAT_MONITOR_LOCK);
                    } catch (Throwable e) {
                        log.error("监控任务执行异常", e);
                    } finally {
                        MDC.clear();
                    }
                },
                30,
                interval,
                TimeUnit.SECONDS,
                false
        );
    }

    @PreDestroy
    public void destroy() {
        isShutdown = true;
        ThreadPoolUtil.shutdownThreadPoolGracefully(executor);
        log.info("监控任务已关闭");
    }

    /**
     * 监控聊天任务
     */
    private void monitorJob() {
        Long lastId = NumberUtils.LONG_ZERO;
        Date currentTime = new Date();
        String todayStr = DateUtil.format(new Date(), DatePattern.NORM_DATETIME_PATTERN) + "&nbsp;&nbsp;&nbsp;";

        List<ChatTimeJob> timeJobList = new ArrayList<>();
        do {
            Stopwatch stopwatch = Stopwatch.createStarted();
            timeJobList = chatTimeJobService.getList(lastId, limit);
            log.info("待处理定时任务数量: {},lastId:{}", timeJobList.size(), lastId);
            if (isShutdown || CollectionUtils.isEmpty(timeJobList)) {
                return;
            }

            Set<Long> ssCfgIdList = new HashSet<>();
            List<String> conversationIdList = new ArrayList<>();
            Set<String> imAccountUserIdList = new HashSet<>();
            Set<Long> csIdList = new HashSet<>();
            timeJobList.forEach(job -> {
                ssCfgIdList.add(job.getSkillSetCfgId());
                conversationIdList.add(job.getConversationId());
                imAccountUserIdList.add(imAccountService.buildUserImUserId(job.getChannelId(), job.getVisitorId()));
                imAccountUserIdList.add(imAccountService.buildCsImUserId(job.getCsId()));
                csIdList.add(job.getCsId());
            });

            // 获取技能组配置信息
            Map<Long, SkillSetConfig> skillSetConfigMap = skillSetConfigService.getSkillSetConfigMap(Lists.newArrayList(ssCfgIdList));
            // 获取会话
            Map<String, Conversation> conversationMap = conversationService.getById(conversationIdList);
            // 获取IM账户信息
            Map<String, Account> imAccountMap = accountService.getMapByUserIds(Lists.newArrayList(imAccountUserIdList));
            // 获取每个会话双方最后发送的聊天消息，key: accountId_conversationId
            Map<String, Message> senderLastMsgMap = messageService.getEverySenderLastMsgBy(conversationIdList);
            // 客服账号状态信息
            Map<Long, CsMemberInfoDTO> csMemberInfoDTOMap = aiCustomerServiceClient.getNormalListBy(Lists.newArrayList(csIdList)).stream().collect(Collectors.toMap(CsMemberInfoDTO::getUserId, v -> v, (o, n) -> o));

            List<ChatTimeJob> updateList = new ArrayList<>();
            List<SendSystemMessageDTO> systemMessageDTOList = new ArrayList<>();
            List<Conversation> closingConversationList = new ArrayList<>();
            Map<String, List<Account>> colsedConversationAccountNotifyMap = new HashMap<>();
            Map<Long, AtomicInteger> releaseCsAllocatedCountMap = new HashMap<>();

            for (ChatTimeJob chatTimeJob : timeJobList) {
                SkillSetConfig skillSetConfig = skillSetConfigMap.get(chatTimeJob.getSkillSetCfgId());
                Conversation conversation = conversationMap.get(chatTimeJob.getConversationId());
                if (skillSetConfig == null || conversation == null || ConversationStateEnum.END.equals(conversation.getState())) {
                    // 无配置信息，删除本次任务；或者会话已结束
                    chatTimeJob.setConversationCloseFlag(1);
                    updateList.add(copyJobForUpdating(chatTimeJob));
                    continue;
                }

                String visitorUserId = imAccountService.buildUserImUserId(chatTimeJob.getChannelId(), chatTimeJob.getVisitorId());
                Account visitorImAccount = imAccountMap.get(visitorUserId);

                String csUserId = imAccountService.buildCsImUserId(chatTimeJob.getCsId());
                Account csImAccount = imAccountMap.get(csUserId);

                if (visitorImAccount == null || csImAccount == null) {
                    log.info("im账号信息缺失，visitorUserId:{},visitorAccount:{}，csUserId:{},csAccount:{}", visitorUserId, visitorImAccount, chatTimeJob.getCsId(), csImAccount);
                    chatTimeJob.setConversationCloseFlag(1);
                    updateList.add(copyJobForUpdating(chatTimeJob));
                    closingConversationList.add(conversation);
                    releaseCsAllocatedCountMap.computeIfAbsent(chatTimeJob.getCsId(), k -> new AtomicInteger()).incrementAndGet();
                    continue;
                }

                if (csMemberInfoDTOMap.get(chatTimeJob.getCsId()) == null) {
                    // 客服账号被冻结或被删除，会话直接结束
                    log.info("客服账号被冻结或被删除，csUserId:{},conversationId:{}", chatTimeJob.getCsId(), conversation.getId());
                    chatTimeJob.setConversationCloseFlag(1);
                    updateList.add(copyJobForUpdating(chatTimeJob));
                    closingConversationList.add(conversation);
                    releaseCsAllocatedCountMap.computeIfAbsent(chatTimeJob.getCsId(), k -> new AtomicInteger()).incrementAndGet();
                    systemMessageDTOList.addAll(createSystemMsg(Lists.newArrayList(visitorUserId, csUserId), conversation, todayStr + skillSetConfig.getEndSessionTip(), 1));
                    continue;
                }

                // 访客最近聊天消息
                Message visitorLastMsg = senderLastMsgMap.get(visitorImAccount.getId() + "_" + chatTimeJob.getConversationId());
                Date visitorLastReplyTime = chatTimeJob.getVisitorLastReplyTime();
                if (visitorLastMsg != null) {
                    visitorLastReplyTime = DateUtil.parse(visitorLastMsg.getCreateTime());
                }
                if (chatTimeJob.getVisitorLastWarnTime() == null || visitorLastReplyTime.after(chatTimeJob.getVisitorLastWarnTime())) {
                    chatTimeJob.setVisitorLastWarnTime(visitorLastReplyTime);
                }

                // 客服最近聊天消息
                Message csLastMsg = senderLastMsgMap.get(csImAccount.getId() + "_" + chatTimeJob.getConversationId());
                Date csLastReplyTime = chatTimeJob.getCsLastReplyTime();
                if (csLastMsg != null) {
                    csLastReplyTime = DateUtil.parse(csLastMsg.getCreateTime());
                }
                if (chatTimeJob.getCsLastWarnTime() == null || csLastReplyTime.after(chatTimeJob.getCsLastWarnTime())) {
                    chatTimeJob.setCsLastWarnTime(csLastReplyTime);
                }

                boolean isUpdateChatTimeJob = false;
                // 开启了访客超时提醒
                if (StatusEnum.ENABLE.getValue().equals(skillSetConfig.getVisitorDisconnTimeoutEnable())) {
                    if (DateUtils.addSeconds(visitorLastReplyTime, skillSetConfig.getVisitorDisconnTimeout()).before(currentTime)
                            && chatTimeJob.getVisitorWarnStatus() == 1) {
                        // 访客断开超时
                        chatTimeJob.setVisitorWarnStatus(2);
                        chatTimeJob.setConversationCloseFlag(1);
                        updateList.add(copyJobForUpdating(chatTimeJob));
                        // 创建系统消息
                        systemMessageDTOList.addAll(createSystemMsg(Lists.newArrayList(visitorUserId, csUserId), conversation, todayStr + skillSetConfig.getEndSessionTip(), 1));
                        // 结束会话
                        closingConversationList.add(conversation);
                        // 推送会话结束通知
                        colsedConversationAccountNotifyMap.computeIfAbsent(conversation.getId(), k -> Lists.newArrayList()).add(csImAccount);
                        // 释放客服被占用的访客进线名额
                        releaseCsAllocatedCountMap.computeIfAbsent(chatTimeJob.getCsId(), k -> new AtomicInteger()).incrementAndGet();
                        continue;
                    } else if (DateUtils.addSeconds(chatTimeJob.getVisitorLastWarnTime(), skillSetConfig.getReplyVisitorTimeout()).before(currentTime)) {
                        // 访客回复超时
                        systemMessageDTOList.addAll(createSystemMsg(Lists.newArrayList(visitorUserId), conversation, todayStr + skillSetConfig.getReplyVisitorTimeoutTip(), 0));
                        chatTimeJob.setVisitorWarnStatus(1);
                        chatTimeJob.setVisitorLastWarnTime(currentTime);
                        isUpdateChatTimeJob = true;
                    }
                }

                // 开启了客服超时提醒
                if (StatusEnum.ENABLE.getValue().equals(skillSetConfig.getCsDisconnTimeoutEnable())) {
                    if (DateUtils.addSeconds(csLastReplyTime, skillSetConfig.getCsDisconnTimeout()).before(currentTime)
                            && chatTimeJob.getCsWarnStatus() == 1) {
                        // 客服断开超时
                        chatTimeJob.setCsWarnStatus(2);
                        chatTimeJob.setConversationCloseFlag(1);
                        updateList.add(copyJobForUpdating(chatTimeJob));
                        // 创建系统消息
                        systemMessageDTOList.addAll(createSystemMsg(Lists.newArrayList(visitorUserId, csUserId), conversation, todayStr + skillSetConfig.getEndSessionTip(), 1));
                        // 结束会话
                        closingConversationList.add(conversation);
                        // 推送会话结束通知
                        colsedConversationAccountNotifyMap.computeIfAbsent(conversation.getId(), k -> Lists.newArrayList()).add(csImAccount);
                        // 释放客服被占用的访客进线名额
                        releaseCsAllocatedCountMap.computeIfAbsent(chatTimeJob.getCsId(), k -> new AtomicInteger()).incrementAndGet();
                        continue;
                    } else if (DateUtils.addSeconds(chatTimeJob.getCsLastWarnTime(), skillSetConfig.getReplyCsTimeout()).before(currentTime)) {
                        // 客服回复超时
                        systemMessageDTOList.addAll(createSystemMsg(Lists.newArrayList(csUserId), conversation, todayStr + skillSetConfig.getReplyCsTimeoutTip(), 0));
                        chatTimeJob.setCsWarnStatus(1);
                        chatTimeJob.setCsLastWarnTime(currentTime);
                        isUpdateChatTimeJob = true;
                    }
                }
                if (!chatTimeJob.getVisitorLastReplyTime().equals(visitorLastReplyTime)
                        || !chatTimeJob.getCsLastReplyTime().equals(csLastReplyTime)) {
                    // 更新最近聊天时间到job记录
                    chatTimeJob.setVisitorLastReplyTime(visitorLastReplyTime);
                    chatTimeJob.setCsLastReplyTime(csLastReplyTime);
                    isUpdateChatTimeJob = true;
                }

                if (isUpdateChatTimeJob) {
                    updateList.add(copyJobForUpdating(chatTimeJob));
                }
            }

            List<String> closingConversationIdList = closingConversationList.stream().map(Conversation::getId).distinct().collect(Collectors.toList());
            log.info("需要更新的任务数量：{}，推送系统消息的数量：{},关闭会话id：{}", updateList.size(), systemMessageDTOList.size(), closingConversationIdList);

            // 结束会话
            conversationService.end(closingConversationIdList);
            // 更新任务信息
            if (CollectionUtils.isNotEmpty(updateList)) {
                chatTimeJobService.updateBatchById(updateList);
            }
            // 释放客服进线被占用的访客名额
            csReceptionLimitService.release(releaseCsAllocatedCountMap);
            // 推送系统消息
            messageService.sendSystemMsg(systemMessageDTOList, imAccountMap);
            // 推送会话结束通知到客服工作台
            conversationService.pushCloseConversationEvent(colsedConversationAccountNotifyMap, closingConversationList);
            // 发送会话归档事件
            archiveEventMqProducerService.sendConversationArchiveEvent(closingConversationIdList);
            // 统计会话时长，用于坐席调度，优化为可通过队列异步化
            conversationStatService.increaseConversationTime(closingConversationList);

            stopwatch.stop();
            log.info("监控任务执行耗时：{}ms", stopwatch.elapsed(TimeUnit.MILLISECONDS));

            lastId = timeJobList.stream().mapToLong(ChatTimeJob::getId).max().orElse(0L);
        } while (CollectionUtils.isNotEmpty(timeJobList) && !isShutdown);
    }

    private ChatTimeJob copyJobForUpdating(ChatTimeJob chatTimeJob) {
        return ChatTimeJob.builder()
                .id(chatTimeJob.getId())
                .conversationCloseFlag(chatTimeJob.getConversationCloseFlag())
                .csLastReplyTime(chatTimeJob.getCsLastReplyTime())
                .csWarnStatus(chatTimeJob.getCsWarnStatus())
                .visitorLastReplyTime(chatTimeJob.getVisitorLastReplyTime())
                .visitorWarnStatus(chatTimeJob.getVisitorWarnStatus())
                .csLastWarnTime(Optional.ofNullable(chatTimeJob.getCsLastWarnTime()).orElseGet(Date::new))
                .visitorLastWarnTime(Optional.ofNullable(chatTimeJob.getVisitorLastWarnTime()).orElseGet(Date::new))
                .build();
    }

    private List<SendSystemMessageDTO> createSystemMsg(List<String> accountUserIdList, Conversation conversation, String tip, Integer systemType) {
        // 与前端约定，系统消息扩展字段带有 systemMsgType=1 的，标识会话结束，systemMsgType=0 表示回复超时
        JSONObject systemMsgExt = new JSONObject();
        systemMsgExt.put("systemMsgType", systemType.toString());
        return accountUserIdList.stream().map(accountUserId -> SendSystemMessageDTO.of(accountUserId, conversation.getId(), tip, systemMsgExt.toJSONString(), conversation.getRoomId())).collect(Collectors.toList());
    }


}
